package com.omuao.message.queue.manager.impl;

import com.omuao.message.queue.core.queue.MessageOutListener;
import com.omuao.message.queue.manager.DelayQueueManager;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * 标准的延迟队列
 *
 * @author omuao
 */
@Repository
public class StandardDelayQueueManagerImpl implements DelayQueueManager {

    public static Logger logger = LoggerFactory.getLogger(StandardDelayQueueManagerImpl.class);

    public static final long NOT_TIMESTAMP = -1L;

    @Resource(name = "delayDb")
    RocksDB db;

    @Resource(name = "delayMessageOutListener")
    MessageOutListener delayMessageOutListener;

    @Override
    public int delete(String id) {
        try {
            db.delete(id.getBytes());
        } catch (RocksDBException e) {
            logger.error(e.getMessage(), e);
        }
        return 1;
    }

    @Override
    public String query(String id) {
        byte[] bytes = new byte[0];
        try {
            bytes = db.get(id.getBytes());
        } catch (RocksDBException e) {
            logger.error(e.getMessage(), e);
        }
        return new String(bytes);
    }

    @Override
    public int insert(String id, String message, Long timestamp) {
        try {
            String body = message.concat(",").concat(timestamp.toString());
            db.put(id.getBytes(), body.getBytes());
        } catch (RocksDBException e) {
            logger.error(e.getMessage(), e);
        }
        return 1;
    }

    @Override
    public int update(String id, String message, Long timestamp) {
        try {
            String body = message.concat(",").concat(timestamp.toString());
            db.put(id.getBytes(), body.getBytes());
        } catch (RocksDBException e) {
            logger.error(e.getMessage(), e);
        }
        return 1;
    }

    @Override
    public void clearTimeoutMessage() {
        RocksIterator iterator = null;
        try {
            iterator = db.newIterator();
            iterator.seekToFirst();
            for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
                byte[] body = iterator.value();

                String bodyMessage = new String(body);

                String[] elements = bodyMessage.split(",");

                if (elements.length == 2) {
                    Long time = System.currentTimeMillis();
                    Long delayTime = Long.parseLong(elements[elements.length-1]);
                    if (delayTime == NOT_TIMESTAMP) {
                        output(elements[0], iterator.key());
                        continue;
                    }
                    if (time < delayTime) {
                        continue;
                    }
                }

                output(elements[0], iterator.key());
            }

        } finally {
            if (iterator != null) {
                iterator.close();
                iterator = null;
            }
        }
    }

    private void output(String element, byte[] key) {
        boolean flag = delayMessageOutListener.output(element);

        if (flag) {
            try {
                db.delete(key);
            } catch (RocksDBException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    @Override
    public void compactRange() {
        try {
            db.compactRange();
        } catch (RocksDBException e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void commit(String message) {
        this.insert(UUID.randomUUID().toString(), message, NOT_TIMESTAMP);
    }

    @Override
    public void commit(String id, String message) {
        this.insert(id, message, NOT_TIMESTAMP);
    }

    @Override
    public void commit(String message, Long timestamp) {
        this.insert(UUID.randomUUID().toString(), message, timestamp);
    }

    @Override
    public void commit(String id, String message, Long timestamp) {
        this.insert(id, message, timestamp);
    }
}
